package com.shujia.flink.source

import org.apache.flink.streaming.api.functions.source.SourceFunction
import org.apache.flink.streaming.api.scala._

object Demo4SourceFunction {

  def main(args: Array[String]): Unit = {


    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    //使用自定义source
    val myDS: DataStream[String] = env.addSource(new MySourceFunction())

    myDS.print()

    env.execute()

  }

}

/**
 * 自定义source
 * 1、实现SourceFunction接口
 * 2、实现里面的run方法，在run方法中读取数据
 * 3、使用ctx将数据发送到下游
 *
 */
class MySourceFunction extends SourceFunction[String] {
  /**
   * run方法由于读取数据的方法，只会执行一次
   *
   * @param ctx :上下文对象，由于将数据发送到下游
   */
  override def run(ctx: SourceFunction.SourceContext[String]): Unit = {

    while (true) {
      //发送数据到下游
      ctx.collect("java")

      Thread.sleep(1000)
    }

  }


  //任务被取消时执行的方法
  override def cancel(): Unit = {
  }
}
